/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.client.Cell;
import com.nhncorp.neptune.client.CellFilter;
import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.TimeLog;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.neptune.tablet.ColumnValue;
import com.nhncorp.neptune.tablet.TabletInfo;

public class TabletMapFile {
  public static final Log LOG = LogFactory.getLog(TabletMapFile.class.getName());
  
  public static final String DATA_FILE = "map.dat";
  public static final String IDX_FILE = "map.idx";

  private String fileId;
  
  protected GPath filePath;
  
  private SortedSet<MapFileIndexRecord> mapFileIndexRecords = new TreeSet<MapFileIndexRecord>();

  private NConfiguration conf;
  
  protected NeptuneFileSystem fs;

  private TabletInfo tabletInfo;
  
  private String columnName;
  
  private int numOfVersion;
  
  //private LocatedBlocks dfsBlockInfos;
  
  public TabletMapFile(
      NConfiguration conf,
      NeptuneFileSystem fs,
      TabletInfo tabletInfo, 
      String columnName,
      String fileId, 
      GPath filePath,
      int numOfVersion) throws IOException {
    this.conf = conf;
    this.fs = fs;
    this.filePath = filePath;
    this.fileId = fileId;
    this.tabletInfo = tabletInfo;
    this.columnName = columnName;
    this.numOfVersion = numOfVersion;
    
    //GPath mapFilePath = new GPath(filePath, DATA_FILE);
    //dfsBlockInfos = fs.getBlockInfos(mapFilePath);
  }

  public TabletMapFile(NConfiguration conf, TabletInfo tabletInfo, String columnName, String fileId, int numOfVersion) throws IOException  {
    this(conf, NeptuneFileSystem.get(conf),
        tabletInfo, 
        columnName, 
        fileId, 
        new GPath(Tablet.getTabletPath(conf, tabletInfo), columnName + "/" + fileId + "/"),
        numOfVersion);
  }

  public TabletInfo getTabletInfo() {
    return tabletInfo;
  }
  
  public String getColumnName() {
    return columnName;
  }
  
  public MapFileWriter getMapFileWriter() throws IOException {
    return new MapFileWriter(fs, conf.getInt("mapfile.indexBlockSize", 128) * 1024);
  }

//  public MapFileReader getMapFileReader() throws IOException {
//    return new MapFileReader(this, new ColumnFilter());
//  }

  public MapFileReader getMapFileReader(Row.Key startRowKey, Row.Key endRowKey, CellFilter cellFilter) throws IOException {
    MapFileReader mapFileReader = new MapFileReader(this, cellFilter);
    mapFileReader.setRange(startRowKey, endRowKey, null);
    return mapFileReader;
  }
  
  public MapFileReader getMapFileReader() throws IOException {
    return new MapFileReader(this, new CellFilter());
  }

  public boolean loadIndex() throws IOException {
    DataInputStream in = null;
    
    GPath idxPath = new GPath(filePath, IDX_FILE);
    GPath dataPath = new GPath(filePath, DATA_FILE);
    try {
      int fileCount = 0;
      if(fs.exists(idxPath))    fileCount++;
      if(fs.exists(dataPath))   fileCount++;
      if(fileCount == 0) {
        return false;
      } else if(fileCount == 1) {
        LOG.error("data or idx file not exists. tabletInfo=" + tabletInfo + ", column=" + columnName + ", fileId=" + fileId);
        return false;
      }
      
      long dataFileLength = fs.getLength(new GPath(filePath, DATA_FILE));
      boolean idxFileError = false;
      long lastIndexOffset = 0;
      
      //LOG.debug("loadIndex:" + idxPath);
      in = new DataInputStream(fs.open(idxPath)); 
      MapFileIndexRecord mapFileIndexRecord = null;
      try {
        while(true) {
          mapFileIndexRecord = new MapFileIndexRecord();
          mapFileIndexRecord.read(in);
          if(mapFileIndexRecord.getOffset() > dataFileLength) {
            idxFileError = true;
            lastIndexOffset = mapFileIndexRecord.getOffset();
            break;
          }        
          mapFileIndexRecords.add(mapFileIndexRecord);
        }
      } catch (EOFException eof) {
        
      }
      
      if(idxFileError) {
        throw new IOException("Can't load map file. index offset(" + lastIndexOffset + 
                                  ") greate than data file length(" + dataFileLength + ").[path=" + filePath + "]");
      }
    } catch (Exception e) {
      e.printStackTrace();
      throw new IOException(e.getMessage());
    } finally {
      if(in != null)  in.close();
    }
    
//    try {
//      synchronized(this) {
//        dfsBlockInfos = fs.getBlockInfos(dataPath);
//      }
//    } catch (Exception e) {
//      e.printStackTrace();
//    }
    return true;
  }
  
  public long getIndexMemorySize() throws IOException {
    long result = 0;
    for(MapFileIndexRecord mapFileIndexRecord: mapFileIndexRecords) {
      result += mapFileIndexRecord.getMemorySize();
    }
    return result;
  }
  
  class MapFileReader implements Scanner, Searchable {
    private Row.Key currentRowKey;
    private DataInputStream dataIn;
    private DataInputStream seekableIn;
    
    private Row.Key startRowKey;
    private Row.Key endRowKey;
    private Cell.Key cellKey;
    private TabletMapFile tabletMapFile;
    
    //private long totalReadTime;
    
    public MapFileReader(TabletMapFile tabletMapFile, CellFilter cellFilter) throws IOException {
      this.tabletMapFile = tabletMapFile;
      GPath mapFilePath = new GPath(filePath, DATA_FILE);
      try {
        seekableIn = fs.openDataInputStream(mapFilePath);
//        seekableIn = fs.openDataInputStream(mapFilePath, dfsBlockInfos);
      } catch (IOException e) {     
        seekableIn = fs.openDataInputStream(mapFilePath);
//        seekableIn = fs.openDataInputStream(mapFilePath, dfsBlockInfos);
      }
    }
    
    public void setRange(Row.Key startRowKey, Row.Key endRowKey, final Cell.Key cellKey) throws IOException {
      this.startRowKey = startRowKey;
      this.endRowKey = endRowKey;
      this.cellKey = cellKey;
      
      Cell.Key scanCellKey = cellKey;
      if(scanCellKey == null) {
        scanCellKey = Cell.Key.EMPTY_KEY;
      }

      MapFileIndexRecord indexRecord = findNearest(startRowKey, scanCellKey);
//      if(indexRecord != null) {
//        LOG.info("indexRecord:" + mapFileIndexRecords.size() + "," + new String(indexRecord.getRowKey().getBytes(), "EUC-KR") + "," + new String(startRowKey.getBytes(), "EUC-KR") + "," + indexRecord);
//      } else {
//        LOG.info("indexRecord:null," + new String(startRowKey.getBytes(), "EUC-KR"));
//      }
      
      if(indexRecord != null) {
        fs.seek(seekableIn, indexRecord.getOffset());
        dataIn = new DataInputStream(new BufferedInputStream(seekableIn));
      } else {
        seekableIn = null;
      }
    }
    
    public ColumnValue next() throws IOException {
      if(dataIn == null) {
        return null;
      }
      
      //long startTime = System.nanoTime();
      MapFileColumnValue mapFileColumnValue = new MapFileColumnValue();
      try {
        mapFileColumnValue.readFields(dataIn);
      } catch (EOFException eof) {
        return null;
      }
      currentRowKey = mapFileColumnValue.getRowKey();
      
      if(currentRowKey.compareTo(endRowKey) > 0) {
        return null;
      }
      
      //index가 가리키는 위치에서 처음 읽은 값이 검색하는 rowKey보다 작은 경우
      //검색 대상 rowKey가 나타날때 까지 이동한다.
//      if("META".equals(tabletInfo.getTableName()) && new String(currentRowKey.getBytes(), "EUC-KR").indexOf("T_TERM") >= 0) {
//        String inputKey = new String(startRowKey.getBytes(), "EUC-KR");
//        LOG.info("Key2222:" + new String(currentRowKey.getBytes(), "EUC-KR") + "," + inputKey + "," + currentRowKey.equals(startRowKey));
//      }
      if(currentRowKey.compareTo(startRowKey) == 0) {
        if(cellKey != null && mapFileColumnValue.getCellKey().compareTo(cellKey) < 0) {
          mapFileColumnValue = skipToSearchTarget();
          if(mapFileColumnValue == null) {
            return null;
          }
        }
      } else if(currentRowKey.compareTo(startRowKey) < 0) {
        currentRowKey = null;
//        try {
//          while(true) {
//            mapFileColumnValue = new MapFileColumnValue();
//            mapFileColumnValue.readFields(dataIn);
//            currentRowKey = mapFileColumnValue.getRowKey();
//            if(currentRowKey.compareTo(startRowKey) == 0) {
//              if(columnKey == null) {
//                break;
//              } else {
//                //columnkey까지 지정된 경우 columnKey와 크거나 같을때 까지 이동
//                if(mapFileColumnValue.getCell.Key().compareTo(columnKey) >= 0) {
//                  break;
//                }
//              }
//            } else if(currentRowKey.compareTo(endRowKey) > 0) {
//              return null;
//            }
//          }
//        } catch (EOFException eof) {
//          
//        }
        mapFileColumnValue = skipToSearchTarget();
        if(mapFileColumnValue == null) {
          return null;
        }
      }
      
      if(currentRowKey == null) {
        return null;
      }
      
      ColumnValue columnValue = new ColumnValue();
      copyColumnValue(mapFileColumnValue, columnValue);
      
      //totalReadTime += (System.nanoTime() - startTime);
      return columnValue;
    }

    private MapFileColumnValue skipToSearchTarget() throws IOException {
      MapFileColumnValue mapFileColumnValue = new MapFileColumnValue();
      try {
        while(true) {
          mapFileColumnValue.readFields(dataIn);
          currentRowKey = mapFileColumnValue.getRowKey();

//          if("META".equals(tabletInfo.getTableName()) && new String(currentRowKey.getBytes(), "EUC-KR").indexOf("T_TERM") >= 0) {
//            String inputKey = new String(startRowKey.getBytes(), "EUC-KR");
//            LOG.info("Key1111:" + new String(currentRowKey.getBytes(), "EUC-KR") + "," + inputKey + "," + currentRowKey.equals(startRowKey));
//          }          
          
          if(currentRowKey.compareTo(startRowKey) >= 0 && currentRowKey.compareTo(endRowKey) <= 0) {
            if(cellKey == null) {
              return mapFileColumnValue;
            } else {
              //columnkey까지 지정된 경우 columnKey와 크거나 같을때 까지 이동
              if(mapFileColumnValue.getCellKey().compareTo(cellKey) >= 0) {
                return mapFileColumnValue;
              }
            }
          } else if(currentRowKey.compareTo(endRowKey) > 0) {
            return null;
          }
        }
      } catch (EOFException eof) {
        
      }      
      return null;
    }
    
    public Row.Key getCurrentRowKey() {
      return currentRowKey;
    }

    public void close() throws IOException {
      if(dataIn != null)  dataIn.close();
      dataIn = null;
      
      //System.out.println("TotalReadTime: " + totalReadTime);
    }

    public int avaliable() throws IOException {
      if(dataIn == null)  return 0;
      return dataIn.available();
    }
  }
  
  class MapFileWriter {
    private DataOutputStream dataOut;
    private DataOutputStream indexOut;  
    private NeptuneFileSystem fs;
    
    private ColumnValue lastColumnRecord;
    
    private int blockWriteBytes = 0;
    private long currentDataFilePos = 0;
    private long indexFileOffset = 0;
    
    //FIXME 성능 테스트 후 사이즈 조정
    private int indexBlockSize;
    private final int BLOCK_RECORD_COUNT = 100;
    
    private int blockWriteRecordCount = 0;
    
    private long sumDataLength = 0;
    private long sumIndexLength = 0;
    
    //private long writeTotalTime = 0;
    public MapFileWriter(NeptuneFileSystem fs, int indexBlockSize) throws IOException {
      //this.indexBlockSize = conf.getInt("mapfile.indexBlockSize", 128) * 1024;
      //fs = NeptuneFileSystem.get(conf);
      this.indexBlockSize = indexBlockSize;
      this.fs = fs;
      fs.mkdirs(filePath);
      //LOG.debug("MapFileWriter open:" + filePath);
      dataOut = new DataOutputStream(new BufferedOutputStream(fs.create(new GPath(filePath, DATA_FILE))));
      indexOut = new DataOutputStream(new BufferedOutputStream(fs.create(new GPath(filePath, IDX_FILE))));      
    }
    
    public void write(ValueCollection columnValues) throws IOException {
      if(columnValues.getValueSize() == 0) {
        return;
      }
      ColumnValue columnValue = columnValues.getWithDeleted();
      
      //long startTime = System.currentTimeMillis();
      int writeSize = columnValues.write(dataOut);
      //writeTotalTime += (System.currentTimeMillis() - startTime);
      sumDataLength += writeSize;
      
      makeFileIndex(writeSize, columnValue, indexBlockSize);
      
      lastColumnRecord = columnValue;
      
    }
    
    public void write(ColumnValue columnValue) throws IOException {
      MapFileColumnValue mapFileColumnValue = new MapFileColumnValue();
      ValueCollection.copyMapFileColumnValue(columnValue, mapFileColumnValue);
      mapFileColumnValue.write(dataOut);
      int writeSize = mapFileColumnValue.size();
      
      makeFileIndex(writeSize, columnValue, indexBlockSize);
      
      lastColumnRecord = columnValue;
      //dataOut.flush();
      //indexOut.flush();
    }

    private void makeFileIndex(int writeSize, ColumnValue columnValue, int checkSize) throws IOException {
      blockWriteBytes += writeSize;
      currentDataFilePos += writeSize;
      blockWriteRecordCount++;
      
      //누적된 writeBytes가 block size(64kb)보다 큰 경우 현재의 레코드를 index에 추가
      if(blockWriteBytes > checkSize || blockWriteRecordCount > BLOCK_RECORD_COUNT) {
        //index 파일에 index 레코드 추가
        MapFileIndexRecord mapFileIndexRecord = new MapFileIndexRecord(columnValue.getRowKey(), columnValue, indexFileOffset);
        mapFileIndexRecord.write(indexOut);
        mapFileIndexRecords.add(mapFileIndexRecord);
        
        sumIndexLength += mapFileIndexRecord.getMemorySize();
        
        blockWriteBytes = 0;
        blockWriteRecordCount = 0;
        indexFileOffset = currentDataFilePos;
      } 
    }
    
    public void close() throws IOException {
      if(blockWriteBytes > 0) {
        makeFileIndex(0, lastColumnRecord, 0);
      }
      
      dataOut.flush();
      indexOut.flush();
            
      IOException exception = null;
      
      if(dataOut != null) {
        try {
          dataOut.close();
        } catch (IOException e) { 
          LOG.error(e.getMessage() + ", sumDataLength=" + sumDataLength, e);
          exception = e;
        }
      }
      if(indexOut != null) {
        try {
          indexOut.close();
        } catch (IOException e) { 
          LOG.error(e.getMessage() + ", sumIndexLength=" + sumIndexLength, e);
          exception = e;
        }
      }
      if(exception != null) {
        throw exception;
      }      
    }
  }

  public ValueCollection scan(Row.Key rowKey, Cell.Key columnKey) throws IOException {
    MapFileReader reader = getMapFileReader();
    try {
      reader.setRange(rowKey, new Row.Key(Constants.MAX_VALUE), columnKey);
  
      ValueCollection valueCollection = new ValueCollection();
      ColumnValue columnValue = null;
      
      while( (columnValue = reader.next()) != null ) {
        Row.Key scanRowKey = columnValue.getRowKey();
        if(scanRowKey.compareTo(rowKey) > 0) {
          break;
        }
        
        if(scanRowKey.equals(rowKey)) {
          if(columnKey == null) {
            valueCollection.add(columnValue, numOfVersion);
          } else {
            if(columnValue.getCellKey().compareTo(columnKey) == 0) {
              valueCollection.add(columnValue, numOfVersion);
            } else if(columnValue.getCellKey().compareTo(columnKey) > 0) {
              break;
            } else {
              
            }
          }
        }
      }
      return valueCollection;
    } finally {
      reader.close();
    }
  }
  

  public List<ColumnValue> scan(Row.Key rowKey) throws IOException {
    MapFileReader reader = null;
    TimeLog.checkPoint();
    try {
      reader = getMapFileReader();
      reader.setRange(rowKey, rowKey, null);

      List<ColumnValue> result = new ArrayList<ColumnValue>();
      ColumnValue columnValue = null;
      
      while ((columnValue = reader.next()) != null) {
        result.add(columnValue);
      }
      //LOG.info("scan size=" + result.size());
      return result;
    } finally {
      if (reader != null) {
        reader.close();
      }
      TimeLog.endCheckPoint("TabletMapFile: Scan Mapfile: " + rowKey);
    }
  }
  
  public boolean hasValue(Row.Key rowKey) throws IOException {
    MapFileReader reader = getMapFileReader();
    try {
      reader.setRange(rowKey, rowKey, null);
      
      ColumnValue columnValue = null;
      
      while( (columnValue = reader.next()) != null ) {
        if(columnValue.getRowKey().compareTo(rowKey) < 0) {
          continue;
        } else if(columnValue.getRowKey().compareTo(rowKey) > 0) {
          return false;
        } else {
          return true;
        }
      }
      return false;
    } finally {
      reader.close();
    }
  }
  
//  public SortedSet<MetaValue> scanClosestMetaColumnValues(Row.Key rowKey, MemorySSTableIF memorySSTable) throws IOException {
//    ValueCollection value = null;
//    
//    MapFileReader reader = getMapFileReader();
//    try {
//      reader.setRange(rowKey, Row.Key.MAX_KEY, null);
//      
//      ColumnValue columnValue = null;
//      
//      Row.Key oldRowKey = null;
//      while( (columnValue = reader.next()) != null ) {
//        if(columnValue.getRowKey().compareTo(rowKey) < 0) {
//          continue;
//        }
//        
//        if(oldRowKey == null) {
//          //value = new ValueCollection(columnValue.getCell.Key());
//          value = new ValueCollection();
//        }
//        if(oldRowKey != null && !oldRowKey.equals(columnValue.getRowKey())) {
//          ColumnValue matchedValue = value.get();
//          if(matchedValue != null && !memorySSTable.isDeleted(matchedValue.getRowKey(), columnName)) {
//            return value.getColumnRecords();
//          }
//          
//          value.clear();
//          //value = new ValueCollection(columnValue.getCell.Key());
//          value = new ValueCollection();
//        }
//        value.add(columnValue, 0);
//        
//        oldRowKey = columnValue.getRowKey();
//      }
//    } finally {
//      reader.close();
//    }
//    
//    if(value != null) {
//      ColumnValue matchedValue = value.get();
//      if(matchedValue != null && !memorySSTable.isDeleted(matchedValue.getRowKey(), columnName)) {
//        return value.getColumnRecords();
//      }
//    }
//    return null;
//  } 
  
  public MapFileIndexRecord findNearest(Row.Key targetRowKey, Cell.Key cellKey) throws IOException {
//    MapFileIndexRecord test = new MapFileIndexRecord(targetRowKey, cellKey);
//    for(MapFileIndexRecord eachRecord: mapFileIndexRecords) {
//      System.out.println(eachRecord.getRowKey() + ">" + eachRecord.compareTo(test));
//    }
    SortedSet<MapFileIndexRecord> tailMap = mapFileIndexRecords.tailSet(new MapFileIndexRecord(targetRowKey, cellKey));
    if(tailMap.size() > 0) {
      return tailMap.first();
    } else {
      return null;
    }
  }

  public GPath getFilePath() {
    return filePath;
  } 
  
  public void moveTo(GPath targetPath, boolean deleteParent) throws IOException {
    GPath originPath = this.filePath;
    GPath originMapDataPath = new GPath(originPath, DATA_FILE);
    GPath originMapIndexPath = new GPath(originPath, IDX_FILE);

    GPath destDataPath = new GPath(targetPath, DATA_FILE);
    GPath destIndexPath = new GPath(targetPath, IDX_FILE);
    
    //LOG.debug("move data file: " + originMapDataPath + " to " + destDataPath);
    if(!fs.move(originMapDataPath, destDataPath)) {
      LOG.error("Can't move data file: " + originMapDataPath + " to " + destDataPath);
      throw new IOException("Can't move data file: " + originMapDataPath + " to " + destDataPath);
    }

    //LOG.debug("move index file: " + originMapIndexPath + " to " + destIndexPath);
    if(!fs.move(originMapIndexPath, destIndexPath)) {
      LOG.error("Can't move index file: " + originMapIndexPath + " to " + destIndexPath);
      throw new IOException("Can't move index file: " + originMapIndexPath + " to " + destIndexPath);
    }

    //delete temp dir
    if(deleteParent) {
      int retry = 0;
      while(retry < 5) {
        //LOG.debug("minor compaction temp dir:" + originPath);
        boolean result = fs.delete(originPath, true);
        if(result) {
          break;
        }
        retry++;
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          return;
        }
      }
      if(retry >= 5) {
        LOG.warn("Can't delete minor compaction temp dir:" + originPath);
      }
      //LOG.debug("moveTo:delete:" + originMapIndexPath + " to " + destIndexPath);
    }
    this.filePath = targetPath;
  }

  public boolean delete() throws IOException {
    int retry = 0;
    while(retry < 5) {
      boolean result = fs.delete(filePath, true);
      if(result) {
        return true;
      }
      try {
        Thread.sleep(500);
      } catch (InterruptedException e) {
      }
    }
    return false;
  } 
  
  public boolean equals(Object obj) {
    if(!(obj instanceof TabletMapFile)) return false;
    
    TabletMapFile otherObj = (TabletMapFile)obj;
    
    
    if(!tabletInfo.getTabletName().equals(otherObj.tabletInfo.getTabletName()))     return false;
    if(!columnName.equals(otherObj.columnName))     return false;
    return fileId.equals(otherObj.fileId);
  }

  public String toString() {
    return "[tabletName=" + tabletInfo.getTabletName() + ", columnName=" + columnName + ", path=" + filePath; 
  }
  
//  public void printIndex() throws IOException {
//    System.out.println("\t\tMapFile: " + filePath);
//    for(MapFileIndexRecord indexRecord: mapFileIndexRecords) {
//      System.out.println("\t\t\t" + indexRecord);
//    }
//  }

  public SortedSet<MapFileIndexRecord> getMapFileIndexRecords() {
    return mapFileIndexRecords;
  }

  public long getDataFileSize() throws IOException {
    return fs.getLength(new GPath(filePath, DATA_FILE));
  }

  /**
   * @return the fileId
   */
  public String getFileId() { 
    return fileId;
  }


  public NeptuneFileSystem getFileSystem() {
    return fs;
  }
  
  public static void copyColumnValue(MapFileColumnValue mapFileColumnValue, ColumnValue columnValue) {
    columnValue.setRowKey(mapFileColumnValue.getRowKey());
    columnValue.setCellKey(mapFileColumnValue.getCellKey());
    columnValue.setValue(mapFileColumnValue.getValue());
    columnValue.setTimestamp(mapFileColumnValue.getTimestamp());
    columnValue.setDeleted(mapFileColumnValue.isDeleted());
  }
  
  public static void main(String[] args) throws IOException {
    if(args.length < 4) {
      System.out.println("Usage: java TabletMapFile <table name> <tablet name> <column name> <file id>");
      System.exit(0);
    }
    
    TabletInfo tabletInfo = new TabletInfo();
    tabletInfo.setTableName(args[0]);
    tabletInfo.setTabletName(args[1]);

    Row.Key rowKey = Tablet.generateMetaRowKey("T_TERA_1T", Row.Key.MAX_KEY); 
    TabletMapFile mapFile = new TabletMapFile(new NConfiguration(), tabletInfo, args[2], args[3], 0);
    
    mapFile.loadIndex();
    
    MapFileReader reader = mapFile.getMapFileReader();
    try {
      reader.setRange(rowKey, rowKey, null);
      ColumnValue columnValue = null;
      while( (columnValue = reader.next()) != null ) {
        System.out.println(">>>>>" + columnValue.getRowKey() + ">" + columnValue.getCellKey() + ">" + columnValue.getTimestamp() + ">" + columnValue.isDeleted());
      }
    } finally {
      reader.close();
    }
  }
}
  
